package com.facebook.presto.tdengine;

import com.facebook.presto.spi.*;
import com.facebook.presto.spi.connector.ConnectorMetadata;
import com.google.common.collect.ImmutableList;

import java.util.*;

/**
 * 该对象定义了需要连接的数据源所有信息（schema，tables，table，column，columnType等）
 *
 * @author heguangchuan
 */

public class TdEngineMetadata implements ConnectorMetadata {

    private String connectId = "tdengine";

    private TdEngineMetadata() {
    }

    private static TdEngineMetadata instance;

    public static TdEngineMetadata getInstance() {
        if (instance == null) {
            synchronized (TdEngineMetadata.class) {
                if (instance == null) {
                    instance = new TdEngineMetadata();
                }
            }
        }
        return instance;
    }

    //该方法对应Presto show schemas命令，用来展示Connector下有哪些数据库
    @Override
    public List<String> listSchemaNames(ConnectorSession session) {
        List<String> dataBases;
        try {
            dataBases = TdUtil.getDataBases();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return dataBases;
    }

    //获取指定表名的表句柄
    @Override
    public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName table) {

        return new TdEngineTableHandle(table.getSchemaName(), table.getTableName(), connectId);
    }

    //返回一个表布局约束
    @Override
    public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns) {

        TdEngineTableHandle tableHandle = (TdEngineTableHandle) table;
        ConnectorTableLayout layout = new ConnectorTableLayout(new TdEngineTableLayoutHandle(tableHandle));
        return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
    }

    @Override
    public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle) {
        return new ConnectorTableLayout(handle);
    }

    @Override
    public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table) {
        TdEngineTableHandle tableHandle = (TdEngineTableHandle) table;
        //todo
        List<ColumnMetadata> list = null;
        try {
            list = TdUtil.getColumnMetadata(tableHandle.getTableName(), tableHandle.getSchemaName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        SchemaTableName tableName = new SchemaTableName(tableHandle.getSchemaName(), tableHandle.getTableName());
        return new ConnectorTableMetadata(tableName, list);
    }

    @Override
    public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle table) {
        TdEngineTableHandle tableHandle = (TdEngineTableHandle) table;
        Map<String, ColumnHandle> map = new HashMap<>();
        List<ColumnMetadata> list = null;
        try {
            list = TdUtil.getColumnMetadata(tableHandle.getTableName(), tableHandle.getSchemaName());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < list.size(); i++) {
            ColumnMetadata meta = list.get(i);
            map.put(meta.getName(), new TdEngineColumnHandle(connectId, meta.getName(), meta.getType(), i));
        }
        return map;
    }

    @Override
    public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle) {
        return ((TdEngineColumnHandle) columnHandle).getColumnMetadata();
    }

    @Override
    public List<SchemaTableName> listTables(ConnectorSession session, Optional<String> schemaName) {
        List<SchemaTableName> listTable = new ArrayList<>();
        try {
            for (String table : TdUtil.getStable(schemaName.get())) {
                listTable.add(new SchemaTableName(schemaName.get(), table));
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return listTable;
    }

    @Override
    public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix) {

        Map<SchemaTableName, List<ColumnMetadata>> columns = new HashMap<>();
        Optional<String> optional = Optional.of(prefix.getSchemaName());
        List<SchemaTableName> list = listTables(session, optional);
        for (SchemaTableName table : list) {
            if (table.getTableName().startsWith(prefix.getTableName())) {
                try {
                    columns.put(table, TdUtil.getColumnMetadata(table.getTableName(), table.getSchemaName()));
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
        return columns;
    }
}
